


/**
 * 设备故障检测(RXJava)时间窗口处理
 * 利用 buff() 函数在限定时间之内新的上送数据 [暂存缓冲区]
 */
private void deviceFaultMonitor(String deviceType, String deviceCode, String jsonStr) {
    // 判断异常信息才会落库
    DeviceMonitorVo deviceMonitor = monitorService.getDeviceMonitor(deviceType, deviceCode, jsonStr);
    if (checkDeviceFault(deviceMonitor)) {
        // 转发处理
        deviceMonitor.setPDeviceType(deviceType);
        deviceMonitor.setDeviceCode(deviceCode);
        Observable.create((ObservableOnSubscribe<DeviceMonitorVo>) observableEmitter -> {
            if (observableEmitter.isDisposed()) {
                return;
            }
            observableEmitter.onNext(deviceMonitor);
        }).buffer(1, TimeUnit.MINUTES)
                .subscribe(new Observer<List<DeviceMonitorVo>>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable disposable) {
                    }

                    @Override
                    public void onNext(@NonNull List<DeviceMonitorVo> dataList) {
                        List<DeviceFaultMonitorDetail> details = new ArrayList<>(dataList.size());
                        dataList.forEach(data -> {
                            // if device status is fault then save to db
                            DeviceFaultMonitorDetail detail = new DeviceFaultMonitorDetail();
                            detail.setDeviceTypeCode(deviceMonitor.getPDeviceType());
                            detail.setDeviceCode(deviceMonitor.getDeviceCode());
                            detail.setDtcCode(deviceMonitor.getDtcCode().toString());
                            detail.setFaultDetail(deviceMonitor.getFaultDesc());
                            String tenantId = getTenantIdByCache(deviceInfoCache.getTenantIdCacheByDeviceCode(), deviceMonitor.getDeviceCode());
                            detail.setTenantId(tenantId);
                            details.add(detail);
                        });
                        if (!CollectionUtils.isEmpty(details)) {
                            monitorDetailService.saveBatch(details);
                        }
                    }

                    @Override
                    public void onError(@NonNull Throwable throwable) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }


}
